package com.batchengine.system;

import static com.batchengine.BatchEngine.getEngine;
import com.batchengine.manager.ScheduledWorkerManager;
import com.batchengine.manager.WorkerManager;
import com.batchengine.task.DefaultTask;
import com.batchengine.task.TaskContext;
import com.batchengine.task.TaskMode;
import static com.batchengine.task.TaskMode.READ;
import com.batchengine.worker.ScheduledTask;
import static com.batchengine.worker.Status.FINISHED;

import java.util.Map;
import java.util.Set;

/**

 * User: sunil.incubator@yahoo.com
 * Date: Sep 13, 2009
 * Time: 6:35:32 PM
 */

/**
 * It cleanups all the scheduled finished tasks
 */
public class ScheduledTaskCleaner extends DefaultTask {

    public ScheduledTaskCleaner() {
        super("ScheduledTaskCleaner");
    }

    public ScheduledTaskCleaner(String name) {
        super("ScheduledTaskCleaner");
    }

    @Override
    public void execute(TaskContext context) {

        Map<Integer, WorkerManager> managers = getEngine().getManagers();
        Set<Integer> set = managers.keySet();
        if (null == set || 0 == set.size()) {
            return;
        }

        for (Integer managerId : set) {

            WorkerManager workerManager = managers.get(managerId);
            if (!(workerManager instanceof ScheduledWorkerManager)) {
                continue;
            }
            ScheduledWorkerManager schManager = (ScheduledWorkerManager) workerManager;
            Map<Long, ScheduledTask> allScheduledTasks = schManager.getAllScheduledTasks();
            if (null == allScheduledTasks || 0 == allScheduledTasks.size()) {
                continue;
            }
            Set<Long> tasks = allScheduledTasks.keySet();
            if (null == tasks || 0 == tasks.size()) {
                continue;
            }
            for (Long aLong : tasks) {
                ScheduledTask scheduledTask = allScheduledTasks.get(aLong);
                if (FINISHED.equals(scheduledTask.getStatus())) {
                    schManager.cleanUpTask(aLong);
                    allScheduledTasks.remove(aLong);
                }
            }
        }
    }

    @Override
    public TaskMode getTaskMode() {
        return READ;
    }
}
